package com.hw.webflux.controller;

import com.hw.webflux.model.User;
import com.hw.webflux.reactivedao.UserReactiveDao;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.r2dbc.core.DatabaseClient;
import org.springframework.data.relational.core.query.Criteria;
import org.springframework.data.relational.core.query.Update;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @program: spring-webflux
 * @description:
 *  1. Mono 表示的是包含 0 或者 1 个元素的异步序列,mono->单一对象 ,user -> Mono<User>
 *  2. Flux 表示的是包含 0 到 N 个元素的异步序列,flux->数组列表对象, List<User> -> Flux<User>
 * @author: hewei
 * @date: 2021-10-29 11:19
 **/
@Slf4j
@RestController
@RequestMapping("/reactive")
public class UserController {

    private static List<String> userList;
    private long delayMillis=1000L;

//    @Resource
//    UserMapper userMapper;
    /*r2dbc 响应式支持*/
    @Resource
    UserReactiveDao userReactiveDao;
    @Resource
    private DatabaseClient databaseClient;

    public UserController() {
        userList=new ArrayList<>();
        userList.add("杨萍");
        userList.add("马国萍");
        userList.add("马青青");
        userList.add("杨溢春");
        userList.add("张港欣");
        userList.add("杨丹");
    }

    /*返回单个对象*/
    @GetMapping("/getUser")
    public Mono<String> getUser(String name){
        String user=userList.stream().filter(m->m.equals(name)).collect(Collectors.toList()).get(0);
        return Mono.just(user);
    }

    /*返回多个对象*/
    @GetMapping("/userList")
    public Flux<Map> userList(String name){
        List<String> users = new ArrayList<>();
        users=userList.stream().filter(m->m.contains(name)).collect(Collectors.toList());
        Map userMap=new HashMap();
        for (String n:users) {
            userMap.put(userMap.size(),n);
        }
        return (Flux<Map>) Flux.fromIterable(userMap.entrySet());
    }

    /**响应式返回，每隔delayMillis毫秒返回一个，直到全部返回，只是一次网络请求
     * 使用h5的sse支持，返回类型"text/event-stream"
     *
     * @return*/
    @GetMapping(value="/userListDelay",produces = {MediaType.TEXT_EVENT_STREAM_VALUE})
    public Flux<Map.Entry<Integer, String>> userListDelay(String name){
        List<String> users = new ArrayList<>();
        users=userList.stream().filter(m->m.contains(name)).collect(Collectors.toList());
        Map<Integer, String> userMap=new HashMap();
        for (String n:users) {
            userMap.put(userMap.size(),n);
        }
        return Flux.fromIterable(userMap.entrySet()).delayElements(Duration.ofMillis(delayMillis));
    }
/*----------------------------------------通过mybatis模拟响应式操作---------------------------------------------*/
//    /**
//     * webflux通过mybatis查询单个用户
//     * */
//    @GetMapping("selectById")
//    public Mono<User> selectById(Integer id){
//         return Mono.fromFuture(
//                 CompletableFuture.supplyAsync(() -> userMapper.selectById(id)));
//    }
//
//    /**
//     * mybatis查询多个用户
//     * webflux间隔1秒返回一条数据，直到全部完成
//     * */
//    @GetMapping(value = "selectByAddr",produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
//    public Flux<User> selectByAddr(String addr) throws ExecutionException, InterruptedException {
////        List<User> users=userMapper.selectByAddr(addr);
//        CompletableFuture<List<User>> future=CompletableFuture.supplyAsync(()->userMapper.selectByAddr(addr));
//         return Flux.fromStream(future.get().stream()
//                 .peek(user -> {
//                     try {
//                         TimeUnit.SECONDS.sleep(1);
//                     } catch (InterruptedException e) {
//                         e.printStackTrace();
//                     }
//                 }));
//    }


/*----------------------------------------r2jdbc的完全响应式操作---------------------------------------------*/
    /**
     * r2jdbc数据源支持的响应式查询
     * */
    @GetMapping("getUserRecById")
    private Mono<User> getUserRecById(Long id){
        return userReactiveDao.findById(id);
    }
    /**
     * r2jdbc数据源支持的响应式查询
     * */
    @GetMapping("findByName")
    private Flux<User> findByName(String name){
        return userReactiveDao.findByName(name);
    }
    /**
     * r2jdbc数据源支持的响应式保存
     *
     * @return*/
    @PostMapping("addUser")
    private Mono<Integer> addUser(@RequestBody User user){
        return databaseClient
                .insert()
                .into("t_user")
                .value("name",user.getName())
                .value("addr",user.getAddr())
                .fetch()
                .rowsUpdated();

        /*通过repository进行操作*/
//        return userReactiveDao.save(user);
    }
    /**
     * r2jdbc数据源支持的响应式更新
     * 通过保存接口可以支持更新，会把id相同的记录进行更新（不够灵活，可以通过databaseClient进行操作）
     *
     * @return*/
    @PostMapping("updateNameById")
    private Mono<Integer> updateNameById(@RequestBody User user){
        return databaseClient
                .update()
                .table("t_user")
                .using(Update
                        .update("name",user.getName())
//                        .set("addr",user.getAddr())
                )
                .matching(Criteria.where("id").is(user.getId()))
                .fetch()
                .rowsUpdated();
//                .subscribe(rows -> System.out.println("更新记录数：" + rows));

        /*通过repository进行操作*/
//        return userReactiveDao.save(user);
    }


}
